package com.clown.hotItemsAnalysis

import java.lang
import java.sql.Timestamp
import java.util.Properties

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.java.tuple.{Tuple, Tuple1}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

// 定义输入数据样例类
case class UserBehavior(userId: Long, itemId: Long, categoryId: Long, behavior: String, timestamp: Long)

// 定义窗口聚合样例类
case class ItemViewCount(itemId: Long, windowEnd: Long, count: Long)

object HotItems {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 定义事件时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // val resource = getClass.getResource("/UserBehavior.csv")
    // val inputStream: DataStream[String] = env.readTextFile(resource.getPath)

    // 从kafka中读取数据
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    // 自定义kafka连接器，消费者
    val consumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("hotitems", new SimpleStringSchema(), properties)
    val inputStream: DataStream[String] = env.addSource(consumer)

    // 从文件中读取数据，并转换成样例类，并提取时间戳生成watermark
    val dataStream: DataStream[UserBehavior] = inputStream
      .map(data => {
        val arr = data.split(",")
        UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toLong, arr(3), arr(4).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)

    // 得到窗口聚合结果
    val aggStream: DataStream[ItemViewCount] = dataStream
      .filter(_.behavior.equals("pv")) // 过滤pv行为
      .keyBy(_.itemId) // 按照商品id分组
      .timeWindow(Time.hours(1), Time.minutes(5)) // 设置滚动窗口进行统计
      .aggregate(new CountAgg(), new ItemViewWindowResult())

    val resultStream = aggStream
      .keyBy(_.windowEnd) // 按照窗口分组，收集当前窗口内的商品count数据
      .process(new TopNHotItem(5)) // 自定义处理流程

    resultStream.print("result")

    env.execute("hot items")
  }
}

// 自定义预聚合函数AggregateFunction，聚合状态就是当前商品的count值
class CountAgg() extends AggregateFunction[UserBehavior, Long, Long] {
  override def createAccumulator(): Long = 0L

  // 每来一条数据，调用一次add，count值加1
  override def add(in: UserBehavior, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1

}

// 自定义窗口函数windowFunction
class ItemViewWindowResult() extends WindowFunction[Long, ItemViewCount, Long, TimeWindow] {
  override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
    val itemId = key
    val windowEnd = window.getEnd
    val count = input.iterator.next()
    out.collect(ItemViewCount(itemId, windowEnd, count))
  }
}

class TopNHotItem(size: Int) extends KeyedProcessFunction[Long, ItemViewCount, String] {
  // 先定义状态:listState
  //  lazy val itemValueCountListState: ListState[ItemViewCount] = getRuntimeContext.getListState(new ListStateDescriptor[ItemViewCount]("itemViewCountList", classOf[ItemViewCount]))
  var itemValueCountListState: ListState[ItemViewCount] = _

  override def open(parameters: Configuration): Unit = {
    itemValueCountListState = getRuntimeContext.getListState(new ListStateDescriptor[ItemViewCount]("itemViewCountList", classOf[ItemViewCount]))
  }

  override def processElement(value: ItemViewCount, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#Context, out: Collector[String]): Unit = {
    // 每来一条数据，直接加入ListState
    itemValueCountListState.add(value)
    // 注册一个windowEnd+1之后触发的定时器
    ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)
  }

  // 当定时器触发，可以认为所有窗口统计结果都已到齐，可以排序输出
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
    println("触发了定时器:" + (timestamp - 1))
    // 为了方便排序，另外定义一个ListBuffer,保存listState中的所有数据
    val allItemViewCounts: ListBuffer[ItemViewCount] = ListBuffer()
    val iterator = itemValueCountListState.get().iterator()
    while (iterator.hasNext) {
      allItemViewCounts += iterator.next()
    }
    // 清空状态
    itemValueCountListState.clear()
    // 排序，按照count大小进行排序，取前N个
    val sortedItemViewCounts = allItemViewCounts
      .sortBy(_.count)(Ordering.Long.reverse)
      .take(size)

    // 将排名信息格式化成String，便于打印输出可视化展示
    val result: StringBuilder = new StringBuilder()
    result.append("窗口结束时间:").append(new Timestamp(timestamp - 1)).append("\n")

    // 遍历结果列表中的每个ItemViewCount，输出到一行
    for (i <- sortedItemViewCounts.indices) {
      val currentItemViewCount = sortedItemViewCounts(i)
      result.append("NO.").append(i + 1).append("：")
        .append("商品ID = ").append(currentItemViewCount.itemId).append("\t")
        .append("热门度 = ").append(currentItemViewCount.count).append("\n")
    }
    result.append("\n===========================================\n\n")

    Thread.sleep(1000)
    out.collect(result.toString())
  }
}